package com.shujia.flink.core

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo4ValueState {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds: DataStream[String] = env.socketTextStream("master", 8888)


    val kvDS: DataStream[(String, Int)] = ds.map(word => (word, 1))


    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)


    /**
      * sum:有状态算子
      *
      * 状态中保存单词的数量
      */

    //val countDS: DataStream[(String, Int)] = keyByDS.sum(1)

    /**
      * process ： flink最底层api,可以操作flink的时间，事件状态
      *
      */

    val countDS: DataStream[(String, Int)] = keyByDS.process(new MyKeyedProcessFunction)

    countDS.print()

    env.execute()

  }

}

class MyKeyedProcessFunction extends KeyedProcessFunction[String, (String, Int), (String, Int)] {


  var state: ValueState[Int] = _

  override def open(parameters: Configuration): Unit = {


    /**
      * 定义状态
      *
      */

    //flink运行时上下文对象
    val context: RuntimeContext = getRuntimeContext

    val valueState = new ValueStateDescriptor[Int]("value", classOf[Int])

    /**
      * ValueState:单值的状态，可以为每一个key保存一个状态
      *
      */
    state = context.getState(valueState)

  }

  override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#Context, out: Collector[(String, Int)]): Unit = {


    //之前的状态
    val last: Int = state.value()


    //当前单词的数量
    val curr: Int = value._2 + last

    //更新状态
    state.update(curr)



    //将数据发送到下游
    out.collect((value._1, curr))


  }
}

